package com.shujia.flink.sql;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Demo01WordCount {
    public static void main(String[] args) throws Exception {
        // 构建Table环境

        // 第一种方式：通过StreamExecutionEnvironment进行构建
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 将DataStream转换成Table进行处理
        DataStreamSource<String> wordDS = env.socketTextStream("master", 8888);

        // 将DataStream注册成表
        tEnv.createTemporaryView("words",wordDS);


        // 执行SQL进行查询
        Table wordsTable = tEnv.sqlQuery("select f0 as word,count(*) as cnt from words group by f0");

        // 打印表结构
        wordsTable.printSchema();

        DataStream<Row> resultDS = tEnv.toChangelogStream(wordsTable);

        resultDS.print();

        env.execute();


    }
}
